#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <pthread.h>

#include "../../msgbus.hpp"

using namespace msgbus;
class Handler: public MsgHandler {
public:
    Handler(): MsgHandler(true) {}
    virtual void handle(const char *topic, const char *data, int data_len) override {
        static int s_count = 0;
        printf("[#%ld] #%d topic %s data %s data_len %d\n", pthread_self(), s_count++, topic, data, data_len);

    }

    virtual bool call(const char *topic, const char *method, const char *data, int data_len, std::string *out_data) override {
        static int s_count = 0;
        if(out_data) {
            *out_data = std::string("call ") + method + " data "+data;
            printf("[#%ld] #%d topic %s method %s ret %s\n", pthread_self(), s_count++, topic, method, out_data->c_str());
            return true;
        }
        return false;
    }
};

int main() {
    Handler *handler = new Handler;
    const char *topic = "aaa/bbb";
    std::string token = MsgBus::inst()->subscribe(topic, handler);
    printf("subscribe token %p %s\n", token.c_str(), token.c_str());
    const char *s = "this is a book";
    for(int i=0; i<10000; i++) {
        char buf[128] = {0};
        snprintf(buf, sizeof(buf)-1, "#%d %s", i, s);
        MsgBus::inst()->publish(topic, buf, strlen(buf), 1000 );
    }
    for(int i=0; i<1000; i++) {
        char buf[128] = {0};
        snprintf(buf, sizeof(buf)-1, "#%d %s", i, s);
        std::string data;
        bool ret = MsgBus::inst()->call(topic, "echo", buf, strlen(buf), &data, 1000);
        if(ret) {
            printf("call return: %ld bytes %s\n", data.size(), data.c_str());
        }
    }
    printf("wait for message\n");
    for(int i=0; i<100; i++) {
        usleep(1000*1000);
    }
    if(!token.empty()) {
        printf("unsubscribe token %s\n", token.c_str());
        MsgBus::inst()->unsubscribe(token.c_str());
    }
    return 1;
}
